 1.Kafka高级特性解析之消费者中消息接收中位移提交
   
   1).Consumer需要向Kafka记录自己的位移数据，这个汇报过程称为提交位移(Committing Offsets)
   2).Consumer 需要为分配给它的每个分区提交各自的位移数据
   3).位移提交的由Consumer端负责的，Kafka只负责保管。__consumer_offsets
   4).位移提交分为自动提交和手动提交
   5).位移提交分为同步提交和异步提交
   (1).自动提交
   Kafka Consumer 后台提交
   开启自动提交： enable.auto.commit=true
   配置自动提交间隔：Consumer端： auto.commit.interval.ms ，默认 5s
   Map<String, Object> configs = new HashMap<>();
   configs.put("bootstrap.servers", "node1:9092");
   configs.put("group.id", "mygrp");
   // 设置偏移量自动提交。自动提交是默认值。这⾥做示例。
   configs.put("enable.auto.commit", "true");
   // 偏移量自动提交的时间间隔
   configs.put("auto.commit.interval.ms", "3000");
   configs.put("key.deserializer", StringDeserializer.class);
   configs.put("value.deserializer", StringDeserializer.class);
   KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs);
   consumer.subscribe(Collections.singleton("tp_demo_01"));
   while (true) {
       ConsumerRecords<String, String> records = consumer.poll(100);
   for (ConsumerRecord<String, String> record : records) {
         System.out.println(record.topic()
           + "\t" + record.partition()
           + "\t" + record.offset()
           + "\t" + record.key()
           + "\t" + record.value());
    }
}
   自动提交位移的顺序：
        配置 enable.auto.commit = true
        Kafka会保证在开始调用poll方法时，提交上次poll返回的所有消息
        因此自动提交不会出现消息丢失，但会重复消费
   重复消费举例
        Consumer 每5s提交 offset
        假设提交 offset 后的3s发生了 Rebalance
        Rebalance 之后的所有 Consumer 从上一次提交的offset处继续消费
        因此Rebalance发生前3s的消息会被重复消费
   (2).异步提交
   使用KafkaConsumer#commitSync()：会提交 KafkaConsumer#poll() 返回的最新 offset
   该方法为同步操作，等待直到 offset 被成功提交才返回
   while (true) {
       ConsumerRecords<String, String> records =
       consumer.poll(Duration.ofSeconds(1));
       process(records); // 处理消息
       try {
            consumer.commitSync();
       } catch (CommitFailedException e) {
            handle(e); // 处理提交失败异常
       }
  }
  commitSync 在处理完所有消息之后
  手动同步提交可以控制offset提交的时机和频率
  手动同步提交会： 
      调用 commitSync 时，Consumer 处于阻塞状态，直到 Broker 返回结果
	  会影响 TPS
	  可以选择拉长提交间隔，但有以下问题：
	      会导致 Consumer 的提交频率下降
		  Consumer 重启后，会有更多的消息被消费
  异步提交
  KafkaConsumer#commitAsync()
  while (true) {
         ConsumerRecords<String, String> records = consumer.poll(3_000);
         process(records); // 处理消息
         consumer.commitAsync((offsets, exception) -> {
           if (exception != null) {
                 handle(exception);
           }
        });
   }
   commitAsync出现问题不会自动重试
   处理方式：
   try {
        while(true) {
        ConsumerRecords<String, String> records =
        consumer.poll(Duration.ofSeconds(1));
        process(records); // 处理消息
        commitAysnc(); // 使用异步提交规避阻塞
       }
   } catch(Exception e) {
        handle(e); // 处理异常
   } finally {
       try {
            consumer.commitSync(); // 最后一次提交使用同步阻塞式提交
      } finally {
            consumer.close();
      }
}
   
   
	  
   
   
   
   
   